package com.rocketmq.consumer.manage.impl;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerOrderly;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.rocketmq.consumer.config.MqConfig;
import com.rocketmq.consumer.exception.ListenerException;
import com.rocketmq.consumer.handle.AbstractListenerHandler;
import com.rocketmq.consumer.manage.ListenerManager;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;

/**
 * Topic监听管理
 *
 * @Author xuziwen
 * @Date 2017-08-21 14:19
 */
@Slf4j
@Component("listenerManager")
public class DefaultListenerManager implements ListenerManager {

    private List<DefaultMQPushConsumer> list = Lists.newArrayList();

    @Autowired
    private ApplicationContext applicationContext;
    @Autowired
    private Map<String, AbstractListenerHandler> map;

    @Autowired
    private MqConfig mqConfig;

    @Override
    public void register() throws Exception {
        Map<String, Object> m = Maps.newHashMap();
        for (Map.Entry<String, AbstractListenerHandler> entry : map.entrySet()) {
            AbstractListenerHandler value = entry.getValue();
            String topicName = value.getTopicName();
            String consumerName = value.getConsumerName();
            String tagsName = value.getTagsName();
            if (m.get(topicName + consumerName) != null) {
                throw new ListenerException("register error topicName:" + topicName
                        + " consumerName:" + consumerName + " beanName:" + entry.getKey());
            }
            m.put(topicName + consumerName, "");
            DefaultMQPushConsumer consumer = buildDefaultMQPushConsumer(topicName, consumerName, tagsName,
                    (MessageListenerOrderly) value , value.getMessageBroadcast());
            list.add(consumer);
            log.info("register successfully topicName:" + topicName +
                    " consumerName:" + consumerName + " tagsName:" + tagsName);
        }
    }

    @Override
    public void unregister() {
        for (DefaultMQPushConsumer consumer : list) {
            consumer.shutdown();
        }
    }

    private DefaultMQPushConsumer buildDefaultMQPushConsumer(
            String topName, String consumerName,
            String tagsName, MessageListenerOrderly bean , boolean messageBroadcast) throws MQClientException {

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerName);
        consumer.setNamesrvAddr(mqConfig.getNamesrvAddr());
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        // 每次拉取10条
        consumer.setConsumeMessageBatchMaxSize(mqConfig.getConsumeMessageBatchMaxSize());
        // consumer与关联的broker保持长连接，默认30s发送心跳
        consumer.setHeartbeatBrokerInterval(mqConfig.getHeartbeatBrokerInterval());
        // consumer轮询拉取消息到本地队列，轮询间隔指定pullInterval，默认0
        consumer.setPullInterval(mqConfig.getPullInterval());
        // consumer本地队列大小，指定pullThresholdForQueue，默认1000
        consumer.setPullThresholdForQueue(mqConfig.getPullThresholdForQueue());
        // 消费进度上报，默认5秒
        consumer.setPersistConsumerOffsetInterval(mqConfig.getPersistConsumerOffsetInterval());

        consumer.subscribe(topName, tagsName);
        consumer.registerMessageListener(bean);

        if(messageBroadcast){
            consumer.setMessageModel(MessageModel.BROADCASTING);
        }
        consumer.start();
        return consumer;
    }

}
